package spider.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import spider.config.Configs;

import java.util.Properties;

public class SendDataToKafka {
	private static Logger logger = LogManager.getLogger(SendDataToKafka.class);
	KafkaProducer<String, String> producer;
	public SendDataToKafka() {

		Properties props = new Properties();
		props.put("metadata.broker.list",Configs.KafkaConfig.METADATA_BROKERS);
		props.put("group.id", Configs.KafkaConfig.GROUP_ID);
        props.put("bootstrap.servers", Configs.KafkaConfig.BOOTSTRAP_SERVERS);
        props.put("enable.auto.commit", Configs.KafkaConfig.ENABLE_AUTO_COMMIT);
        props.put("metadata.broker.list", Configs.KafkaConfig.METADATA_BROKERS);
        props.put("auto.offset.reset", Configs.KafkaConfig.AUTO_OFFSET_RESET);
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		producer = new KafkaProducer<String, String>(props);
	}

	public void send(String topic,String data) {
		//logger.info("send to kafka topic is ---->" + topic +" data is --> " + data);
		producer.send(new ProducerRecord<String, String>(topic, data));
	}

}
